package org.dhy.elasticsearch.interfaces;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.util.TypeUtils;
import org.dhy.elasticsearch.ElasticConfigProperties.ElasticConfig;
import org.dhy.elasticsearch.enums.Logical;
import org.dhy.elasticsearch.processer.EsEntityProcesser;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @author longye
 */
public class AbstractEsRepository<T> implements EsRepository<T>, InitializingBean {
    private EsEntityProcesser<T> esEntityProcesser;
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    @Autowired
    private ElasticConfig elasticConfig;

    private String realIndexName;

    @Override
    public void afterPropertiesSet() {
        ParameterizedType parameterizedType = (ParameterizedType) getClass().getGenericSuperclass();
        Class<T> tClass = (Class<T>) parameterizedType.getActualTypeArguments()[0];
        esEntityProcesser = new EsEntityProcesser<>(tClass);
        checkIndex();
    }

    private QueryBuilder baseSearchBuilder(T t) throws NoSuchFieldException, IllegalAccessException {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

        final List<EsEntityProcesser.SearchField> esFields = esEntityProcesser.getEsFields();
        for (EsEntityProcesser.SearchField searchField : esFields) {
            if (searchField.getSearchAble()) {
                final Object fieldValue = getFieldValue(searchField.getFieldName(), t);
                final MatchQueryBuilder message = QueryBuilders.matchQuery(searchField.getFieldName(), fieldValue);
                if (searchField.getLogical().equals(Logical.AND)) {
                    boolQuery.must(message);
                } else {
                    boolQuery.should(message);
                }
            }
        }

        return boolQuery;
    }

    @Override
    public List<T> baseSearch(T o, Integer size) throws IOException, NoSuchFieldException, IllegalAccessException {
        SearchSourceBuilder searchBuilder = getSearchSourceBuilder(o);
        Assert.notNull(size, "请提供搜索的大小");
        searchBuilder.size(size);

        SearchRequest request = new SearchRequest(this.realIndexName);
        request.source(searchBuilder);
        return search(request);
    }

    @Override
    public List<T> basePagedSearch(T o, Integer pageSize, Integer pageNum) throws IOException, NoSuchFieldException, IllegalAccessException {
        SearchSourceBuilder searchBuilder = getSearchSourceBuilder(o);
        searchBuilder.from((pageNum - 1) * pageSize).size(pageSize);

        SearchRequest request = new SearchRequest(this.realIndexName);
        request.source(searchBuilder);
        return search(request);
    }

    private SearchSourceBuilder getSearchSourceBuilder(T o) throws NoSuchFieldException, IllegalAccessException {
        QueryBuilder query = baseSearchBuilder(o);
        final EsEntityProcesser.Sort sort = esEntityProcesser.getSort();
        FieldSortBuilder order = SortBuilders.fieldSort(sort.getFieldName()).order(sort.getSortOrder());
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.timeout(TimeValue.timeValueMinutes(2L));
        searchBuilder.query(query);
        searchBuilder.sort(order);
        return searchBuilder;
    }

    public List<T> search(SearchRequest request) throws IOException {
        SearchResponse response = this.restHighLevelClient.search(request, RequestOptions.DEFAULT);
        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            System.out.println("部分分片副本处理失败！");
            for (ShardSearchFailure failure : response.getShardFailures()) {
                String reason = failure.reason();
                System.out.println("分片处理失败原因：" + reason);
            }
        }
        final List<T> ts = parseSearchResponse(response);
        return ts;
    }

    @Override
    public T findById(String id) {
        GetRequest request = new GetRequest(this.realIndexName, id);
        // 实时(否)
        request.realtime(false);
        // 检索之前执行刷新(是)
        request.refresh(true);

        GetResponse response = null;
        try {
            response = restHighLevelClient.get(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            System.out.println("查找失败！");
        }

        if (Objects.nonNull(response)) {
            if (response.isExists()) {
                final String sourceAsString = response.getSourceAsString();
                return JSONObject.parseObject(sourceAsString, this.esEntityProcesser.getRealClass());
            } else {
                // 处理未找到文档的方案。 请注意，虽然返回的响应具有404状态代码，但仍返回有效的GetResponse而不是抛出异常。
                // 此时此类响应不持有任何源文档，并且其isExists方法返回false。
                System.out.println("文档未找到，请检查参数！");
            }
        }
        return null;
    }

    @Override
    public void checkIndex() {
        try {
            final EsEntityProcesser.Index index = esEntityProcesser.getIndex();
            final String defaultIndex = elasticConfig.getDefaultIndex();

            if (EsEntityProcesser.Index.checkIndexNoNull(index)) {
                this.realIndexName = index.getName();
                if (!checkIndexExists(index.getName())) {
                    createIndex(index.getName(), index.getReplicas(), index.getShards());
                }
                return;
            }

            if (!StringUtils.isEmpty(defaultIndex) && !checkIndexExists(defaultIndex)) {
                createIndex(defaultIndex, 1, 1);
                return;
            }

            System.out.println("请在Document注解上指定索引名称或者在配置文件中配置DefaultIndex");
        } catch (IOException e) {
            System.out.println("创建索引失败，detail：" + e.getMessage());
        }
    }

    @Override
    public void insert(T t) {
        if (!checkIndexExists(this.realIndexName)) {
            checkIndex();
        }
        try {
            final Object id = getFieldValue(this.esEntityProcesser.getIdField(), t);
            IndexRequest request = new IndexRequest(this.realIndexName).id(id.toString()).source(JSON.toJSONString(t), XContentType.JSON);
            // request的opType默认是INDEX(传入相同id会覆盖原document，CREATE则会将旧的删除)
            // request.opType(DocWriteRequest.OpType.CREATE)
            IndexResponse response = null;
            response = restHighLevelClient.index(request, RequestOptions.DEFAULT);

            String index = response.getIndex();
            String documentId = response.getId();
            if (response.getResult() == DocWriteResponse.Result.CREATED) {
                System.out.println("新增文档成功！");
            } else if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                System.out.println("修改文档成功！");
            }
            // 分片处理信息
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                System.out.println("文档未写入全部分片副本");
            }
            // 如果有分片副本失败，可以获得失败原因信息
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    System.out.println("副本失败原因" + reason);
                }
            }
        } catch (IOException | NoSuchFieldException | IllegalAccessException e) {
            System.out.println("文档新增失败！:" + e.getMessage());
        }
    }

    private Object getFieldValue(String fieldName, T t) throws NoSuchFieldException, IllegalAccessException {
        final Field field = t.getClass().getField(fieldName);
        field.setAccessible(true);
        return field.get(t);
    }

    @Override
    public void update(T t) {
        try {
            final Object id = getFieldValue(this.esEntityProcesser.getIdField(), t);
            UpdateRequest request = new UpdateRequest(this.realIndexName, id.toString());
            request.doc(JSON.toJSONString(t), XContentType.JSON);
            // 如果要更新的文档不存在，则根据传入的参数新建一个文档
            request.docAsUpsert(true);
            UpdateResponse response = restHighLevelClient.update(request, RequestOptions.DEFAULT);
            if (response.getResult() == DocWriteResponse.Result.CREATED) {
                System.out.println("文档新增成功！");
            } else if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                System.out.println("文档更新成功！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                System.out.println("文档已被删除，无法更新！");
            } else if (response.getResult() == DocWriteResponse.Result.NOOP) {
                System.out.println("操作没有被执行！");
            }

            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                System.out.println("分片副本未全部处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    System.out.println("未处理原因：" + reason);
                }
            }
        } catch (IllegalAccessException | NoSuchFieldException | IOException e) {
            System.out.println("更新失败！:" + e.getMessage());
        }
    }

    @Override
    public void bulkAdd(List<T> list) {
        try {
            BulkRequest bulkRequest = new BulkRequest();
            for (T dataMap : list) {
                final String id = getFieldValue(this.esEntityProcesser.getIdField(), dataMap).toString();
                String jsonString = JSON.toJSONString(dataMap);
                if (!StringUtils.isEmpty(id)) {
                    IndexRequest request = new IndexRequest(this.realIndexName).id(id).source(jsonString, XContentType.JSON);
                    bulkRequest.add(request);
                }
            }
            // 超时时间(2分钟)
            bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
            // 刷新策略
            bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

            if (bulkRequest.numberOfActions() == 0) {
                System.out.printf("参数错误，批量增加操作失败！");
                return;
            }
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            // 全部操作成功
            if (!bulkResponse.hasFailures()) {
                System.out.println("批量增加操作成功！");
            } else {
                for (BulkItemResponse bulkItemResponse : bulkResponse) {
                    if (bulkItemResponse.isFailed()) {
                        BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                        System.out.println("增加失败详情: " + failure.getMessage());
                    } else {
                        System.out.println("文档增加成功");
                    }
                }
            }
        } catch (NoSuchFieldException | IllegalAccessException | IOException e) {
            System.out.println("增量新增错误：" + e.getMessage());
        }
    }

    @Override
    public void bulkUpdate(List<T> list) {
        try {
            BulkRequest bulkRequest = new BulkRequest();
            for (T dataMap : list) {
                final String id = getFieldValue(this.esEntityProcesser.getIdField(), dataMap).toString();
                String jsonString = JSON.toJSONString(dataMap);
                if (!StringUtils.isEmpty(id)) {
                    UpdateRequest request = new UpdateRequest(this.realIndexName, id).doc(jsonString, XContentType.JSON);
                    bulkRequest.add(request);
                }
            }
            if (bulkRequest.numberOfActions() == 0) {
                System.out.println("参数错误，批量更新操作失败！");
                return;
            }
            bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
            bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (!bulkResponse.hasFailures()) {
                System.out.println("批量更新操作成功！");
            } else {
                for (BulkItemResponse bulkItemResponse : bulkResponse) {
                    if (bulkItemResponse.isFailed()) {
                        BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                        System.out.println("文档更新失败！");
                        System.out.println("更新失败详情: " + failure.getMessage());
                    } else {
                        System.out.println("文档更新成功！");
                    }
                }
            }
        } catch (NoSuchFieldException | IllegalAccessException | IOException e) {
            System.out.println("批量更新失败：" + e.getMessage());
        }
    }

    @Override
    public void delete(String id) {
        DeleteRequest request = new DeleteRequest(this.realIndexName, id);
        DeleteResponse response = null;
        try {
            response = restHighLevelClient.delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            System.out.println("删除失败!");
        }
        if (Objects.nonNull(response)) {
            if (response.getResult() == DocWriteResponse.Result.NOT_FOUND) {
                System.out.println("不存在该文档，请检查参数！");
            }
            System.out.println("文档已删除！");
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                System.out.println("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    System.out.println("失败原因：" + reason);
                }
            }
        }
    }


    /**
     * 判断索引是否存在
     *
     * @param indexName
     * @return
     * @throws IOException
     */
    private boolean checkIndexExists(String indexName) {
        GetIndexRequest request = new GetIndexRequest(indexName);
        try {
            return restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            System.out.println("判断索引是否存在，操作异常！");
        }
        return false;
    }

    /**
     * 创建索引
     *
     * @param indexName
     * @param replicas
     * @param shards
     * @throws IOException
     */
    private void createIndex(String indexName, Integer replicas, Integer shards) throws IOException {
        if (checkIndexExists(indexName)) {
            System.out.println(indexName + "索引已经存在！");
            return;
        }
        Settings.Builder builder = Settings.builder()
                .put("index.number_of_shards", shards)
                .put("index.number_of_replicas", replicas);
        CreateIndexRequest request = new CreateIndexRequest(indexName).settings(builder);
        request.mapping(esEntityProcesser.generateBuilder());
        CreateIndexResponse response = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
        // 指示是否所有节点都已确认请求
        boolean acknowledged = response.isAcknowledged();
        // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            System.out.println("创建索引成功！索引名称为" + indexName);
        }
    }

    private List<T> parseSearchResponse(SearchResponse response) {
        List<T> resultList = new ArrayList<>();
        SearchHit[] hits = response.getHits().getHits();
        for (SearchHit hit : hits) {
            final String sourceAsString = hit.getSourceAsString();
            resultList.add(JSONObject.parseObject(sourceAsString, this.esEntityProcesser.getRealClass()));
        }
        return resultList;
    }

}
